package util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

public final class Util {

    private static long time2stamp(String time) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = sdf.parse(time);
        return date.getTime();
    }

    public static void putTable(Connection connection, TableName tableName, Put[] puts)
            throws IOException {
        Admin admin = connection.getAdmin();
        if (!admin.tableExists(tableName)) {
            TableDescriptorBuilder builder =
                    TableDescriptorBuilder.newBuilder(tableName);
            builder.setColumnFamily(
                    ColumnFamilyDescriptorBuilder.newBuilder(
                            "data".getBytes(StandardCharsets.UTF_8)
                    ).build());
            TableDescriptor td = builder.build();
            admin.createTable(td);
        }
        Table table = connection.getTable(tableName);
        for (Put put : puts) {
            table.put(put);
        }
        table.close();
    }

    private static void addColumn(JSONArray value, Put put) {
        for (int i = 0; i < value.size(); i++) {
            JSONObject object = value.getJSONObject(i);
            for (String key : object.keySet()) {
                put.addColumn(
                        Bytes.toBytes("data"),
                        Bytes.toBytes(String.format("%s_%d", key, i)),
                        Bytes.toBytes(object.getString(key)));
            }
        }
    }

    private static Put monitorType(String topic, JSONObject rawData) throws ParseException {
        JSONObject data = rawData.getJSONObject("LIST");
        String type = data.getString("MONITORTYPE");
        String time = String.valueOf(time2stamp(data.getString("TIME")));
        String rowKey = topic + '_' + time;
        JSONArray value = data.getJSONArray("DATALIST");
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(
                Bytes.toBytes("data"),
                Bytes.toBytes("MONITORTYPE"),
                Bytes.toBytes(type));
        put.addColumn(
                Bytes.toBytes("data"),
                Bytes.toBytes("time"),
                Bytes.toBytes(time));
        addColumn(value, put);
        return put;
    }

    private static Put[] ambientType(String topic, JSONObject rawData) throws ParseException {
        JSONArray array = rawData.getJSONArray("LIST");
        Put[] puts = new Put[array.size()];
        for (int i = 0; i < array.size(); i++) {
            JSONObject data = array.getJSONObject(i);
            String time = String.valueOf(time2stamp(data.getString("TIME")));
            String ambient = data.getString("RA_AMBIENTTYPE");
            String rowKey = topic + "_" + System.currentTimeMillis() + "_" + i;
            JSONArray value = data.getJSONArray("DATALIST");
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(
                    Bytes.toBytes("data"),
                    Bytes.toBytes("RA_AMBIENTTYPE"),
                    Bytes.toBytes(ambient));
            put.addColumn(
                    Bytes.toBytes("data"),
                    Bytes.toBytes("TIME"),
                    Bytes.toBytes(time));
            addColumn(value, put);
            puts[i] = put;
        }
        return puts;
    }

    private static Put robotCircle(String topic, JSONObject rawData) {
        JSONObject data = rawData.getJSONObject("contentValue");
        String time = String.valueOf(System.currentTimeMillis());
        String rowKey = topic + '_' + time;
        JSONArray value = data.getJSONArray("data");
        Put put = new Put(Bytes.toBytes(rowKey));
        addColumn(value, put);
        System.out.println();
        return put;
    }

    public static Put[] getPut(String topic, JSONObject rawData) throws ParseException {
        switch (topic) {
            case "device.9a7709b9f09544c3aaa4eeb77559be71.6c9c97eec2a7458ab834d5d75bb20771.iot.event.monitortype_report":
                return new Put[]{monitorType(topic, rawData)};
            case "device.9a7709b9f09544c3aaa4eeb77559be71.6c9c97eec2a7458ab834d5d75bb20771.iot.event.ra_ambienttype_report":
                return ambientType(topic, rawData);
            case "device.f1885d3e933847819f9f57b5b775a3c4.1435822106.iot.event.robot_circle_detection":
                return new Put[]{robotCircle(topic, rawData)};
            default:
                return null;
        }
    }

}
